package com.pw.study.flink.app

import com.alibaba.fastjson.JSONObject
import org.apache.flink.streaming.api.windowing.triggers.{Trigger, TriggerResult}
import org.apache.flink.streaming.api.windowing.windows.TimeWindow

/**
 * @Desc: trigger
 * @author: future
 * @since: 2022/6/9-21:02
 */
class MyTrigger extends Trigger[JSONObject, TimeWindow] {
  override def onElement(t: JSONObject, l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
    println(s"onElement == $l  w: ${w.getStart}==> ${w.getEnd} mark: ${triggerContext.getCurrentWatermark}")
    println(s"t==>$t")
    TriggerResult.CONTINUE
  }

  override def onProcessingTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
    println(s" ===onProcessingTime == $l  w: ${w.getStart}==> ${w.getEnd} mark: ${triggerContext.getCurrentWatermark}")
    TriggerResult.CONTINUE
  }

  override def onEventTime(l: Long, w: TimeWindow, triggerContext: Trigger.TriggerContext): TriggerResult = {
    println(s" ===onEventTime == $l  w: ${w.getStart}==> ${w.getEnd} mark: ${triggerContext.getCurrentWatermark}")

    TriggerResult.FIRE
  }

  override def clear(w: TimeWindow, triggerContext: Trigger.TriggerContext): Unit = {
    println("clear=============================")
  }
}
